package cn.wangjie.spark.udf

import org.apache.spark.sql.expressions.UserDefinedFunction
import org.apache.spark.sql.{DataFrame, SparkSession}

/**
 * SparkSQL中定义UDF函数，分别在SQL和DSL中使用
 */
object SparkSQLUdf {

  def main(args: Array[String]): Unit = {
		
		// 1. 创建SparkSession实例对象，通过建造者模式
		val spark: SparkSession = SparkSession.builder()
			// 设置应用名称
			.appName(this.getClass.getSimpleName.stripSuffix("$"))
			// 设置运行模式
			.master("local[2]")
			.getOrCreate()
		// 导入隐式转换函数
		import spark.implicits._
		
		
		// 加载JSON格式数据
		val empDF: DataFrame = spark.read.json("datas/resources/employees.json")
		
		// TODO: 方式一：在SQL中使用
		spark.udf.register(
			"lower_name", // 函数名称
			(name: String) => {  // 函数体
				name.toLowerCase
			}
		)
		
		// 注册为临时视图
		empDF.createOrReplaceTempView("view_tmp_emps")
		spark.sql(
			"""
			  |SELECT name, lower_name(name) AS new_name FROM view_tmp_emps
			  |""".stripMargin)
    			.show()
		
		println("=========================================================")
		
		// TODO: 方式一：在DSL中使用
		import org.apache.spark.sql.functions.udf
		val lower_udf: UserDefinedFunction = udf(
			(name: String) => {  // 函数体
				name.toLowerCase
			}
		)
		
		empDF
			.select(
			$"name", lower_udf($"name").as("new_name")
			)
			.show()
		
		// 应用结束，关闭资源
		spark.stop()
	}
	
}
